package com.lzcu.maoyan.mapreduce.mapreduce;

import com.lzcu.maoyan.mapreduce.mapreduce.mapper.StockMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.StockReducer;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.StockWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;

/**
 * @author WangLiang
 * @date `2024/4/24`
 */
public class PhoenixMR extends Configured implements Tool {

    @Override
    public int run(String[] strings) throws Exception {
        //1.创建job
//            Configuration conf = getConf();
//            conf.set(HConstants.ZOOKEEPER_QUORUM,"hbase");
//            conf= HBaseConfiguration.create(conf);
        Configuration conf = new Configuration();
        conf.addResource("phoenix/core-site.xml");
        conf.addResource("phoenix/hbase-site.xml");
        conf.addResource("phoenix/hdfs-site.xml");
        Job job = Job.getInstance(conf, "PhoenixStockApplication");
        job.setJarByClass(PhoenixMR.class);

        //2.设置输入输出格式
        job.setInputFormatClass(PhoenixInputFormat.class);
        job.setOutputFormatClass(PhoenixOutputFormat.class);

        //3.设置数据读入和写出路径
        String selectQuery = "SELECT STOCK_NAME,RECORDING_YEAR,RECORDINGS_QUARTER FROM STOCK ";
        PhoenixMapReduceUtil.setInput(job, StockWritable.class, "STOCK", selectQuery);
        PhoenixMapReduceUtil.setOutput(job, "STOCK_STATS", "STOCK_NAME,MAX_RECORDING");

        //4.设置代码片段
        job.setMapperClass(StockMapper.class);
        job.setReducerClass(StockReducer.class);

        //5.设置Mapper和Reducer端的输出key,value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(StockWritable.class);

        TableMapReduceUtil.addDependencyJars(job);

        //6.任务提交
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new PhoenixMR(), args);
    }

}
